package cn.com.netedge.iot.frontend.comm.task;

import ch.qos.logback.core.Appender;
import cn.com.netedge.iot.frontend.comm.constants.AppConstant;
import cn.com.netedge.iot.frontend.comm.constants.ChannelConstant;
import cn.com.netedge.iot.frontend.comm.utils.DeviceLogUtil;
import cn.com.netedge.iot.frontend.comm.utils.FeCommCacheUtil;
import cn.com.netedge.iot.frontend.comm.utils.JmsMsgUtil;
import cn.com.netedge.iot.frontend.comm.vo.DelayMessage;
import com.alibaba.fastjson2.JSON;
import cn.com.netedge.iot.common.constants.CacheKeyConstants;
import cn.com.netedge.iot.common.constants.MqConstants;
import cn.com.netedge.iot.common.dataobject.DeviceRunningInfo;
import cn.com.netedge.iot.common.dataobject.SendFrame;
import cn.com.netedge.iot.common.enums.MessageDirectEnum;
import cn.com.netedge.iot.common.enums.OnlineStatusEnum;
import cn.com.netedge.iot.common.utils.DateUtil;
import cn.com.netedge.iot.common.utils.RedisUtil;
import cn.com.netedge.iot.common.utils.StringUtil;
import cn.com.netedge.iot.model.online.DataCOnlineModel;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;

@Component
public class FixDelayTask {
    private static Logger logger = LoggerFactory.getLogger(FixDelayTask.class);
    private final static int LENGTH_MIN = 12;

    private static final int MAX_CONCURRENT_CONNECTION = 10000; //单台前置支持最多的并发连接数

    private static final int HEART_BEAT_INTERVAL = 30 * 1000; // 30 秒
    private static final int LOGFILE_CHECK_INTERVAL = 3 * 60 * 1000; //3分钟
    private static final int LOGFILE_MAX_LIVE_TIME = 15 * 60 * 1000; //15分钟

    private static final int CHANNEL_CHECK_INTERVAL = 30 * 1000; //30 秒
    private static final int CHANNEL_MAX_LIVE_TIME = 60 * 1000; //1分钟

    private final static int DEVICE_MESSAGE_SEND_TIME_OUT = 60 * 2; //2分钟前的报文直接丢弃，不再发送
    private static final int OFFLINE = 0;
    private static int OFFLINE_THRESHOLD = 180;



    @Value("${frontEnd.deviceMaxConnection}")
    private String maxConnection;

    @Value("${frontEnd.offlineThreshold}")
    private String offlineThreshold;

    @PostConstruct
    public void init() {
        if(StringUtil.isNotEmpty(offlineThreshold) && StringUtil.isDigits(offlineThreshold)) {
            OFFLINE_THRESHOLD = Integer.parseInt(offlineThreshold);
        }
    }

    @Scheduled(fixedDelay = 60 * 1000) //1分钟
    public void syncDeviceStatus() {
        if (!isLeader()) {
//            logger.info("{} is not leader, can't exec sysncRtuStatus", BpConstants.machineName);
            return;
        }

        Set<String> keys = RedisUtil.hkeys(CacheKeyConstants.KEY_DEVICE_RUNNING + "map");
        if (keys.isEmpty()) {
            logger.info("device_running*'s keys is empty.");
            return;
        }
        DeviceRunningInfo runningInfo;
        String json;
        for (String key : keys) {
            json = RedisUtil.get(key);
            if (StringUtil.isEmpty(json)) {
                logger.info("key:{} 's cache json is empty.", key);
                FeCommCacheUtil.delFilterKey(key);
                continue;
            }
            runningInfo = JSON.parseObject(json, DeviceRunningInfo.class);
            if (runningInfo == null) {
                logger.info("key:{} 's cache json：{} convert error.", key, json);
                FeCommCacheUtil.delFilterKey(key);
                continue;
            }
            Date lastOnlineTime = runningInfo.getOnlineTime();

            int curStatus = runningInfo.getCurStatus();
            String deviceNumber = runningInfo.getDeviceNumber();
            int deviceType = runningInfo.getDeviceType();
            //最后在线时间 距离当前时间 超过5分钟，被认为是离线状态
            if (OnlineStatusEnum.ONLINE.getCode() == curStatus && lastOnlineTime != null) {
                Date now = new Date();
                Date compareTime = DateUtil.addSeconds(lastOnlineTime, OFFLINE_THRESHOLD);
                if (now.after(compareTime)) {
                    logger.info("deviceNumber:{} deviceType:{} is offline.", deviceNumber, deviceType);
                    //缓存中也将状态更新为离线
                    runningInfo.setCurStatus(OnlineStatusEnum.OFFLINE.getCode());
                    FeCommCacheUtil.setString(key, JSON.toJSONString(runningInfo));

                    pushOfflineEvent(runningInfo);
                }
            }
            runningInfo.setOldStatus(curStatus);
            runningInfo.setOldOnlineTime(lastOnlineTime);
            FeCommCacheUtil.setString(key, JSON.toJSONString(runningInfo));
        }
    }

    private void pushOfflineEvent(DeviceRunningInfo runningInfo) {
        DataCOnlineModel dataCOnlineModel = new DataCOnlineModel();
        dataCOnlineModel.setDeviceId(runningInfo.getDeviceId());
        dataCOnlineModel.setDeviceType(runningInfo.getDeviceType());
        dataCOnlineModel.setIsOnline(OFFLINE);
        Date now = new Date();
        dataCOnlineModel.setOccurTime(DateUtil.DateToLongStr(now));
        JmsMsgUtil.sendMessage(MqConstants.QUEUE_DATA_C_ONLINE_MESSAGE, JSON.toJSONString(dataCOnlineModel));
    }

    @Scheduled(fixedDelay = 30 * 1000)
    public void cleanChannel() {
        if (!isLeader()) {
            logger.info("{} is not leader", AppConstant.machineName);
            return;
        }

        Set<String> aliveChildren = RedisUtil.hkeys(CacheKeyConstants.KEY_FRONTEND_COMM_MACHINE + "machine_map");
        if (aliveChildren == null || aliveChildren.isEmpty()) {
            logger.info("{} 's name list is empty.", CacheKeyConstants.KEY_FRONTEND_COMM_MACHINE);
            return;
        }

        Set<String> deviceSet = RedisUtil.hkeys(CacheKeyConstants.KEY_FRONTEND_COMM_MACHINE + "device_map");
        if (deviceSet == null || deviceSet.isEmpty()) {
            return;
        }
        String machine, channelId, deviceNumber, runInfo, key;
        DeviceRunningInfo deviceRunningInfo;
        int index, lastIndex, deviceType;
        boolean isAlive = false;
        for (String device : deviceSet) {
            if (StringUtil.isEmpty(device) || device.length() < 5) {
                FeCommCacheUtil.delKey(device);
                continue;
            }
            index = device.indexOf("/");
            if (index < 0) {
                logger.debug("/ index < 0");
                FeCommCacheUtil.delKey(device);
                continue;
            }
            machine = device.substring(0, index);
            isAlive = false;
            for (String child : aliveChildren) {
                if (child.equalsIgnoreCase(machine)) { //该应用还存活着
                    isAlive = true;
                    break;
                }
            }
            if (isAlive) {
                continue;
            }
            //查找已消亡采集前置的终端通道信息
            channelId = RedisUtil.get(device);
            if (StringUtil.isEmpty(channelId)) {
                continue;
            }
            //删除已废弃的通道信息
            key = CacheKeyConstants.KEY_CHANNEL + channelId;
            RedisUtil.delete(key);
            key = CacheKeyConstants.KEY_DEVICE_INFO + channelId;
            RedisUtil.delete(key);
            FeCommCacheUtil.delKey(device);

            //清除终端任务信息
            if (device.length() <= index) {
                continue;
            }
            lastIndex = device.lastIndexOf("/");
            if(index == lastIndex) {
                continue;
            }
            deviceNumber = device.substring(index + 1, lastIndex);
            deviceType = Integer.parseInt(device.substring(lastIndex + 1));
            key = CacheKeyConstants.KEY_DEVICE_TASK + deviceNumber + "_" + deviceType;
            RedisUtil.delete(key);
            //更新终端运行信息
            key = CacheKeyConstants.KEY_DEVICE_RUNNING + deviceNumber + "_" + deviceType;
            runInfo = RedisUtil.get(key);
            if (StringUtil.isEmpty(runInfo)) {
                continue;
            }
            deviceRunningInfo = JSON.parseObject(runInfo, DeviceRunningInfo.class);
            if (deviceRunningInfo == null) {
                continue;
            }
            deviceRunningInfo.setCurStatus(OnlineStatusEnum.OFFLINE.getCode());
            deviceRunningInfo.setMachineName(null);
            FeCommCacheUtil.setString(key, JSON.toJSONString(deviceRunningInfo));
        }
    }

    @Scheduled(fixedDelay = HEART_BEAT_INTERVAL)
    public void hearBeat() {
        if (StringUtil.isEmpty(AppConstant.machineName)) {
            logger.info("machine name is empty.");
            return;
        }
        Date now = new Date();
        RedisUtil.hset(CacheKeyConstants.KEY_FRONTEND_COMM_MACHINE + "machine_map", AppConstant.machineName, DateUtil.DateToLongStr(now));
    }

    /**
     * 判断自己是否为领导，最小序号节点名与机器名是否相同
     *
     * @return
     */
    private static boolean isLeader() {
        String key = CacheKeyConstants.KEY_FRONTEND_COMM_MACHINE + "machine_map";
        Set<String> machineNameList = RedisUtil.hkeys(key);
        if (machineNameList == null || machineNameList.isEmpty()) {
            logger.info("{} 's name list is empty.", key);
            return false;
        }
        List<String> children = new ArrayList<>(machineNameList.size());
        Date now = new Date();
        Date compareTime = DateUtil.addMilliseconds(now, -(HEART_BEAT_INTERVAL * 3));
        String value;
        Date hearBeatTime;
        for (String machineName : machineNameList) {
            value = RedisUtil.hget(key, machineName);
            if (StringUtil.isEmpty(value)) {
                logger.info("machine:{} 's heartbeat time is null.", machineName);
                continue;
            }
            hearBeatTime = DateUtil.LongStrToDate(value);
            if (hearBeatTime == null) {
                logger.info("value:{} can't convert date.", value);
                continue;
            }
            if (hearBeatTime.after(compareTime)) { //离线
                logger.debug("machine:{} --> online.", machineName);
                children.add(machineName);
                continue;
            }
            RedisUtil.hdel(key, machineName);
        }
        if (children.size() > 0) {
            Collections.sort(children);
            String leader = children.get(0);
            if (leader == null || !leader.equals(AppConstant.machineName)) {
                return false;
            }
        }
        return true;
    }

    @Scheduled(fixedDelay = LOGFILE_CHECK_INTERVAL)
    public void checkLogFile() {
        logger.debug("enter check log file.");
        long now = System.currentTimeMillis();
        Iterator<Map.Entry<String, Long>> entryIt = DeviceLogUtil.loggerLiveMap.entrySet().iterator();
        Map.Entry<String, Long> entry;
        Long lastLogTime;
        String key;
        ch.qos.logback.classic.Logger deviceLogger;
        Appender appender;
        List<String> removeKeyList = new LinkedList<>();
        while (entryIt.hasNext()) {
            entry = entryIt.next();
            lastLogTime = entry.getValue();
            if (lastLogTime == null) {
                continue;
            }
            if ((now - lastLogTime) < LOGFILE_MAX_LIVE_TIME) { //最后记录日志时间距今未超过 指定分钟数
                continue;
            }
            key = entry.getKey();
            deviceLogger = DeviceLogUtil.loggerMap.get(key);
            if (deviceLogger == null) {
                logger.info("{}'s rtuLogger is null", key);
                continue;
            }

            try {
                deviceLogger.detachAndStopAllAppenders();
                logger.debug("{} logger close.", key);
            } catch (Exception e) {
                logger.debug("close appender error.", e);
            }
            removeKeyList.add(key);
        }

        for (String remveKey : removeKeyList) {
            DeviceLogUtil.loggerMap.remove(remveKey);
            DeviceLogUtil.loggerLiveMap.remove(remveKey);
        }
    }

    @Scheduled(fixedDelay = CHANNEL_CHECK_INTERVAL)
    public void checkChannel() {
        logger.debug("enter check channel.");
        long now = System.currentTimeMillis();
        Iterator<Map.Entry<String, Long>> entryIt = ChannelConstant.CHANNEL_CREATE_TIME_MAP.entrySet().iterator();
        Map.Entry<String, Long> entry;
        Long lastLogTime;
        String key;
        Channel channel;
        List<String> removeKeyList = new LinkedList<>();
        while (entryIt.hasNext()) {
            entry = entryIt.next();
            lastLogTime = entry.getValue();
            if (lastLogTime == null) {
                continue;
            }
            if ((now - lastLogTime) < CHANNEL_MAX_LIVE_TIME) {
                continue;
            }
            key = entry.getKey();
            channel = ChannelConstant.CHANNEL_MAP.get(key);
            if (channel == null) {
                logger.info("{}'s channel is null", key);
                continue;
            }

            try {
                channel.disconnect();
                logger.debug("{} channel disconnect.", key);
            } catch (Exception e) {
                logger.debug("disconnect channel error.", e);
            }
            removeKeyList.add(key);
        }

        for (String remveKey : removeKeyList) {
            ChannelConstant.CHANNEL_CREATE_TIME_MAP.remove(remveKey);
            ChannelConstant.CHANNEL_MAP.remove(remveKey);
        }

        //并发连接数超限，主动端口最早的连接，允许其他设备连接
        int maxConnectionInt = MAX_CONCURRENT_CONNECTION;
        if (!StringUtil.isEmpty(maxConnection) && StringUtil.isDigits(maxConnection)) {
            maxConnectionInt = Integer.parseInt(maxConnection);
        }
        int channelCount = ChannelConstant.CHANNEL_MAP.size();
        int batchKillConnection = channelCount - (maxConnectionInt * 4 / 5);
        if (channelCount > MAX_CONCURRENT_CONNECTION) {
            logger.debug("concurrent_connections_num:{} > max_concurrent_connection:{} ", channelCount, MAX_CONCURRENT_CONNECTION);
            Iterator<Map.Entry<String, Channel>> channelEntryIt = ChannelConstant.CHANNEL_MAP.entrySet().iterator();
            Map.Entry<String, Channel> channelEntry;

            int count = 0;
            removeKeyList.clear();
            while (channelEntryIt.hasNext() && count < batchKillConnection) {
                count++;
                channelEntry = channelEntryIt.next();
                key = channelEntry.getKey();
                removeKeyList.add(key);
                channel = channelEntry.getValue();
                if (channel == null) {
                    logger.info("{}'s channel is null", key);
                    continue;
                }

                try {
                    channel.disconnect();
                    logger.debug("{} channel disconnect.", key);
                } catch (Exception e) {
                    logger.debug("disconnect channel error.", e);
                }
            }
            for (String remveKey : removeKeyList) {
                ChannelConstant.CHANNEL_MAP.remove(remveKey);
            }
        }

    }

    @Scheduled(fixedDelay = 3 * 1000)
    public void sendRtuDelayMessage() {
        if (AppConstant.rtuDelayMessageMap.isEmpty()) {
            return;
        }
        Iterator<Map.Entry<String, List<DelayMessage>>> entryIterator = AppConstant.rtuDelayMessageMap.entrySet().iterator();
        Map.Entry<String, List<DelayMessage>> entry;
        List<DelayMessage> messageList;
        DelayMessage delayMessage = null;
        Long lastSendTime;
        String commAddr, channelId, path;
        long now;
        Channel channel;
        while (entryIterator.hasNext()) {
            try {
                entry = entryIterator.next();
                commAddr = entry.getKey();
                messageList = entry.getValue();
                if (messageList.isEmpty()) {
                    logger.info("comm_addr:{} message list is empty.", commAddr);
                    AppConstant.rtuDelayMessageMap.remove(commAddr);
                    continue;
                }
                lastSendTime = ChannelConstant.SEND_FLAG_MAP.get(commAddr);
                now = System.currentTimeMillis();
                if (lastSendTime != null && (now - lastSendTime) / 1000 < AppConstant.MESSAGE_SEND_TIMEOUT) {
                    logger.info("comm_addr:{} is waiting, already exe {} s", commAddr, (now - lastSendTime) / 1000);
                    continue;
                }
                path = AppConstant.machineName + "/" + commAddr;
                channelId = RedisUtil.get(path);
                if (channelId == null) {
                    logger.warn("channelId:{} 缓存中通道信息不存在", path);
                    continue;
                }
                channel = ChannelConstant.CHANNEL_MAP.get(channelId);
                if (channel == null || !channel.isActive()) {
                    logger.info("channelId:{} 通道非活动状态", channelId);
                    continue;
                }
                int count = 0;
                while (count <= 0 && messageList.size() > 0) {
                    delayMessage = messageList.get(0);

                    SendFrame frame = toFrame(delayMessage.getMessage());
                    if ((now - delayMessage.getCreateTime()) / 1000 > DEVICE_MESSAGE_SEND_TIME_OUT) {
                        logger.debug("报文:{} 是{} 秒前的，直接丢弃", delayMessage.getMessage(), DEVICE_MESSAGE_SEND_TIME_OUT);
                        messageList.remove(0);
                        continue;
                    }
                    count++;
                    sendMessage(path, channelId, channel, frame);
                    messageList.remove(0);
                }
            } catch (Exception e) {
                logger.warn("send delay message error.", e);
            }
        }
    }

    private SendFrame toFrame(String message) {
        if (StringUtil.isEmpty(message)) {
            logger.error("messsage is null");
            return null;
        }
        logger.debug("message text: {}", message);
        return JSON.parseObject(message, SendFrame.class);

    }

    private void sendMessage(String path, String channelId, Channel channel, SendFrame frame) {
        if (frame == null) {
            logger.error("frame is null");
            return;
        }
        byte[] buf = frame.getData();
        if (buf == null) {
            logger.error("frame's data is null");
            return;
        }
        if (buf.length < LENGTH_MIN) {
            logger.error("frame's data length is not enough");
            return;
        }
        channel.writeAndFlush(buf);
        ChannelConstant.SEND_FLAG_MAP.put(frame.getDeviceNumber() + "_" + frame.getDeviceType(), System.currentTimeMillis());
        //记录报文日志
        try {
            DeviceLogUtil.deviceLogger.logMessage(frame.getDeviceNumber(), frame.getDeviceType(), MessageDirectEnum.DELAY_DOWN, frame.getData());
        } catch (Exception e) {
            logger.info("record rtu log error.", e);
        }
    }
}
